package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	cluster "github.com/bsm/sarama-cluster"
	"log"
	"os"
	"os/signal"
	"sync"
)

var wg sync.WaitGroup
func main()  {


}
//集群操作
func lusterConsumer(group sync.WaitGroup,brokers,topics []string, groupId string)  {
	defer wg.Done()
	config := cluster.NewConfig()
	config.Consumer.Return.Errors =true
	config.Group.Return.Notifications =true
	config.Consumer.Offsets.Initial =sarama.OffsetNewest
	consumer, err := cluster.NewConsumer(brokers,groupId,topics,config)
	if err !=nil {
		log.Panic(err)
		return
	}
	defer  consumer.Close()
	signals :=  make(chan os.Signal,1)
	signal.Notify(signals,os.Interrupt)

	// consume errors
	go func() {
		for err := range consumer.Errors(){
			log.Printf("%s:Error: %s\n", groupId, err.Error())
		}
	}()
	// consume notifications
	go func() {
		for ntf := range consumer.Notifications(){
			log.Printf("%s:Rebalanced: %+v \n", groupId, ntf)
		}
	}()
	// consume messages, watch signals
	var successes int
	//标签组
	LOOP:
		for  {
			select {
			case msg,ok:= <-consumer.Messages():
				if ok {
					fmt.Fprintf(os.Stdout, "%s:%s/%d/%d\t%s\t%s\n", groupId, msg.Topic, msg.Partition, msg.Offset, msg.Key, msg.Value)
					consumer.MarkOffset(msg, "")  // mark message as processed
					successes++
				}
			case <-signals:
				break LOOP
			}
		}
	fmt.Fprintf(os.Stdout, "%s consume %d messages \n", groupId, successes)

}